/**    
* @Title: OkexWebsocketApi.java
* @Package: org.vdalf.gsgz.api.impl
* @author vDalf 2020-09-26 11:30:38 上午11:30:38
* @Copyright © 2020 vDalf Block-Chain Foundation (In Preparation Non-Governmental Organization) All rights reserved.
*/
package org.vdalf.gsgz.api.impl;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import org.apache.commons.compress.compressors.deflate64.Deflate64CompressorInputStream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.vdalf.gsgz.config.OkexRobotProperties;
import org.vdalf.gsgz.service.DataCenter;
import org.vdalf.gsgz.utils.MarketUtils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;

import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import okhttp3.WebSocket;
import okhttp3.WebSocketListener;
import okio.ByteString;

/**    
* @Title: OkexWebsocketApi.java
* @Package: org.vdalf.gsgz.api.impl
* @Description: 
* @author vDalf 2020-09-26 11:30:38 上午11:30:38
*/
@Service("okexWebsocketApi")
public class OkexWebsocketApi implements IWebsocketApi{
	
	private static final Logger logger = LoggerFactory.getLogger(OkexWebsocketApi.class);
	
	private static final int PING_TIME = 25_000;
	
	private static final int PONG_TIMEOUT = 3_000;
	
	private WebSocket ws;
	
	private long time;
	
	private Set<String> channels = new HashSet<>();
	
	@Autowired
	private OkexRobotProperties okexRobotProperties;

	@Override
	public void open() {
		logger.info("okex.ws.open....................");
		if(!okexRobotProperties.getServer().isEnableWebsocket()) {
			logger.info("okex行情配置不可用....");
			return;
		}
		try {
			OkHttpClient client = new OkHttpClient();
			Request request = new Request.Builder().url(okexRobotProperties.getServer().getWsUrl()).build();
			client.newWebSocket(request, new WebSocketListener() {
				
				@SuppressWarnings("serial")
				@Override
				public void onOpen(WebSocket webSocket, Response response) {
					ws = webSocket;
					time = System.currentTimeMillis();
					if(!channels.isEmpty()) {
						JSONObject json = new JSONObject();
						json.put("op", "subscribe");
						json.put("args", new JSONArray() {{channels.stream().forEach(this::add);}});
						logger.info("ws.send({})",json);
						webSocket.send(json.toJSONString());
					}
				}
				
				@Override
				public void onMessage(WebSocket webSocket, String text) {
					time = System.currentTimeMillis();
					logger.info("text-plain:{}",text);
				}
				
				@Override
				public void onMessage(WebSocket webSocket, ByteString bytes) {
					try {
						time = System.currentTimeMillis();
						String message = uncompress(bytes.toByteArray());
						if(okexRobotProperties.getServer().isEnableWebsocketLogger()) {
							logger.info("\r\nbyte-array:{}",message);
						}
						if("pong".equals(message)) {
							return;
						}
						JSONObject json = JSON.parseObject(message);
//            		byte-array:{"table":"spot/ticker","data":[{"last":"236.36","open_24h":"242.76","best_bid":"236.29","high_24h":"244.02","low_24h":"229.64","base_volume_24h":"272262.4602","quote_volume_24h":"64525634.4536","best_ask":"236.3","instrument_id":"BCH-USDT","timestamp":"2020-01-09T13:45:37.494Z","best_bid_size":"21","best_ask_size":"10.662","last_qty":"0.1101"}]}
						if(json.getString("table") != null) {
							json.getJSONArray("data").stream().filter(c->channels.stream().anyMatch(ch->ch.split(":")[1].equals(((JSONObject)c).getString("instrument_id")))).forEach(c->{
								DataCenter.getInstance().updatePrice(((JSONObject)c).getString("instrument_id"), ((JSONObject)c).getBigDecimal("last"));
							});
						}
					} catch (Exception e) {
						logger.error("推送message解析失败....");
					}
				}
			});
			ScheduledExecutorService pool = Executors.newSingleThreadScheduledExecutor();
			pool.scheduleAtFixedRate(new Runnable() {
				@Override
				public void run() {
					try {
						String ping = "ping";
						if(System.currentTimeMillis() - time > PING_TIME) {
							logger.info("长时间未收到数据..开始发送ping....");
							ws.send(ping);
							synchronized (ping) {
								try {
									ping.wait(PONG_TIMEOUT);
								} catch (InterruptedException e) {
									logger.error("e---->",e);
								}
							}
							if(System.currentTimeMillis() - time > PING_TIME) {
								logger.error("长时间未收到pong..重启ws连接....");
								ws.close(1000, null);
								ws.cancel();
								OkexWebsocketApi.this.open();
								pool.shutdownNow();
							}
						}
					} catch (Exception e) {
						logger.error("ping error:",e);
					}
				}
			}, 3_000, 5_000, TimeUnit.MILLISECONDS);
			
		} catch (Throwable e) {
			logger.error("websocket open error:",e);
		}
	}

	
	
	public void initChannels(Set<String> markets) {
		logger.info("init channels : markets = {}",markets);
		this.channels = markets.stream().map(c->{
			if(MarketUtils.isFuture(c)) {
				return "futures/ticker:"+c;
			}else if(MarketUtils.isSwap(c)) {
				return "swap/ticker:"+c;
			}else 
				return "spot/ticker:"+c;
		}).collect(Collectors.toSet());
		logger.info("channels inited : {}",this.channels);
	}

	@SuppressWarnings("serial")
	@Override
	public void subscribe(String market) {
		String channel;
		if(MarketUtils.isFuture(market)) {
			channel = "futures/ticker:"+market;
		}else if(MarketUtils.isSwap(market)) {
			channel = "swap/ticker:"+market;
		}else 
			channel = "spot/ticker:"+market;
		channels.add(channel);
		JSONObject json = new JSONObject();
		json.put("op", "subscribe");
		json.put("args", new JSONArray() {{channels.stream().forEach(this::add);}});
		logger.info("ws.send({})",json);
		ws.send(json.toJSONString());
	}

	private static String uncompress(byte[] bytes) {
        try (final ByteArrayOutputStream out = new ByteArrayOutputStream();
             final ByteArrayInputStream in = new ByteArrayInputStream(bytes);
             final Deflate64CompressorInputStream zin = new Deflate64CompressorInputStream(in)) {
            final byte[] buffer = new byte[1024];
            int offset;
            while (-1 != (offset = zin.read(buffer))) {
                out.write(buffer, 0, offset);
            }
            return out.toString();
        } catch (final IOException e) {
            throw new RuntimeException(e);
        }
    }
	
}
